import { Service } from "egg";
import { Server as SocketIOServer } from "socket.io";
import { createServer } from "http";
import * as Y from "yjs";
import { Awareness } from "y-protocols/awareness";
// 端口 被占用
// lsof -nP -iTCP:7001 -sTCP:LISTEN
// kill -9 <占用7001的PID>

// lsof -nP -iTCP:7002 -sTCP:LISTEN
// kill -9 <占用7002的PID>

// 房间上下文：管理每个房间的 Y.Doc 与 Awareness 状态（服务端内存）
type RoomContext = {
  doc: Y.Doc;
  awareness: Awareness;
};


// 心跳检测和重连核心原理（以 socket.io/engine.io 为例）
// 心跳机制（服务器发 ping，客户端回 pong）
// 服务器按 pingInterval 定时发送 ping。
// 客户端立即回 pong；若服务器在 pingTimeout 内收不到 pong，则认为连接失活并主动断开。
// 断开原因常见为 “ping timeout”，随后触发客户端的重连流程。
// 断开与重连流程
// 断开触发：网络中断、服务端关闭、跨层负载均衡中断、心跳超时等。
// 客户端重连：由 socket.io 的 Manager 统一调度，按退避策略（reconnectionDelay/Max、随机抖动、最大次数 reconnectionAttempts）发起 connect 尝试，事件顺序通常为：
// reconnect_attempt(n) → 重试第 n 次
// reconnect_error(err) → 单次失败
// reconnect(n) → 成功
// reconnect_failed → 超过最大次数仍失败
// 首次连接失败走 connect_error；连接后掉线再连走上面的重连事件。
export default class SocketService extends Service {
  private io: SocketIOServer | null = null;

  /** 从 Cookie 请求头解析指定名称的 Cookie 值（兜底） */
  private getCookieValueFromHeader(cookieHeader: string, name: string): string | null {
    if (!cookieHeader) return null;
    const pairs = cookieHeader
      .split(";")
      .map((c: string) => c.trim())
      .filter(Boolean);
    for (const p of pairs) {
      const idx = p.indexOf("=");
      const k = idx > -1 ? p.slice(0, idx) : p;
      if (k === name) {
        const v = idx > -1 ? decodeURIComponent(p.slice(idx + 1)) : "";
        return v || null;
      }
    }
    return null;
  }

  /** 统一获取明文 SID（优先 Egg cookies，兜底原始 header 手动解析） */
  private getSessionIdFromSocket(socket: any): string | null {
    const cookieHeader = (socket.handshake.headers as any)?.cookie || "";
    const anonCtx = this.app.createAnonymousContext({ headers: { cookie: cookieHeader } } as any);
    const fromEgg = anonCtx.cookies.get("SSO_SESSION_ID", { signed: false });
    if (fromEgg) return fromEgg;
    const fromHeader = this.getCookieValueFromHeader(cookieHeader, "SSO_SESSION_ID");
    return fromHeader;
  }
  
  /** 通过 SID 校验会话得到 userId（成功返回字符串，失败返回 null） */
  private async getUserIdBySessionId(sessionId: string): Promise<string | null> {
    try {
      const anonCtx = this.app.createAnonymousContext();
      const result = await (anonCtx.service as any).userSequelize.validateSession(sessionId);
      if (result?.success && result?.data?.user?.id) return String(result.data.user.id);
      return null;
    } catch {
      return null;
    }
  }

  /** 统一计算某用户在某文档（房间）下的权限 */
  private async computeRoomPermission(documentId: string, currentUserId: string | null) {
    const { models } = this.app as any;
    const result = { effectivePermission: "read" as "read" | "write" | "admin", canWrite: false, isOwner: false };
    try {
      const dbDoc = await models.Document.findOne({ where: { id: documentId, status: 1 } });
      const isOwner = Boolean(dbDoc && currentUserId && String(dbDoc.owner_id) === String(currentUserId));
      result.isOwner = isOwner;
      if (isOwner) {
        result.effectivePermission = "admin";
        result.canWrite = true;
        return result;
      }
      if (currentUserId) {
        const collab = await models.DocumentCollaboration.findOne({ where: { document_id: documentId, user_id: currentUserId } });
        const p = (collab?.permission ?? null) as null | "read" | "write" | "admin";
        if (p === "admin") { result.effectivePermission = "admin"; result.canWrite = true; }
        else if (p === "write") { result.effectivePermission = "write"; result.canWrite = true; }
        else if (p === "read") { result.effectivePermission = "read"; result.canWrite = false; }
      }
    } catch (err) {
      this.app.logger.warn("[socket] compute permission failed room=%s err=%s", documentId, String(err));
    }
    return result;
  }

  public async init() {
    // 避免重复初始化（以 app 级单例为准）
    const appAny = this.app as any;
    if (appAny.__socketIO) return (this.io = appAny.__socketIO);
    if (this.io) {
      appAny.__socketIO = this.io;
      return this.io;
    }
    const server = createServer();
    const rooms: Map<string, RoomContext> = new Map();

    // 性能优化：为每个房间维护一个“增量缓冲区”，定时把缓冲合并后一次性落库
    // 背景：客户端在输入时会产生大量细粒度的 Y 更新，如果每条都写数据库，
    // 会造成数据库写入频繁、锁竞争严重，进而引发服务卡顿甚至崩溃。
    // 为啥不能采用 节流？因为这样会导致 数据延迟 不同步

    // 方案：服务端在内存中实时应用到 Y.Doc（保持权威状态最新），
    // 但对数据库采用“窗口聚合”策略：
    //   - 固定窗口：每 FLUSH_INTERVAL_MS 毫秒把当前 Doc 编码为快照一次 save；

    //   - 阈值触发：当缓冲条数 >= MAX_UPDATES_BEFORE_FLUSH 时立即 save 一次；
    // 如此将 N 次写合并为 1 次，显著降低 DB 压力。

    // 每个 doc 有自己的 updates 增量缓冲区，有自己的 timer ：延迟落库任务。
    const updateBuffers: Map<string, { updates: Uint8Array[]; timer: NodeJS.Timeout | null; lastFlush: number; flushing: boolean }>
      = new Map();


    const FLUSH_INTERVAL_MS = 400;  // 聚合窗口大小（可按机器性能和并发情况调节）
    const MAX_UPDATES_BEFORE_FLUSH = 150; // 缓冲条数阈值（到达后立即落库）

    // 短窗口合并广播：避免高并发下逐条广播造成的事件风暴（40ms 人眼无感）
    const broadcastBuffers: Map<string, { updates: Uint8Array[]; timer: NodeJS.Timeout | null }>
      = new Map();
    const BROADCAST_COALESCE_MS = 40;

    // 性能优化：权限短期缓存（减少频繁数据库查询）
    const permissionCache: Map<string, { value: { effectivePermission: "read" | "write" | "admin"; canWrite: boolean; isOwner: boolean }; ts: number }>
      = new Map();
    const PERM_TTL_MS = 5_000; // 5s 短期缓存：写权限通常不会在秒级频繁变化

    // 从缓存获取用户在房间的权限；未命中或过期则查询并回填缓存
    const getPermissionCached = async (roomName: string, uid: string | null) => {
      const key = `${roomName}:${uid || "(null)"}`;
      const now = Date.now();
      const cached = permissionCache.get(key);
      if (cached && now - cached.ts < PERM_TTL_MS) return cached.value;
      const value = await this.computeRoomPermission(roomName, uid);
      permissionCache.set(key, { value, ts: now });
      return value;
    };

    // 安排一次“延迟落库”任务。如果已经安排过，则复用现有计时器
    const scheduleFlush = (roomName: string, ctx: RoomContext) => {
      let buf = updateBuffers.get(roomName);
      if (!buf) {
        buf = { updates: [], timer: null, lastFlush: 0, flushing: false };
        updateBuffers.set(roomName, buf);
      }

      if (buf.timer) return; // 已经安排，无需重复设置

      buf.timer = setTimeout(async () => {
        // 到点执行一次批量落库
        const target = updateBuffers.get(roomName);

        if (!target || target.flushing || target.updates.length === 0) {
          if (target) target.timer = null;
          return;
        }
        //  落库任务 正在执行
        target.flushing = true;
        try {
          // 用“当前内存 Doc 的完整状态”生成快照一次保存，

          // 取代窗口内多次增量写库，避免 DB 高频 IO
          const snapshot = Y.encodeStateAsUpdate(ctx.doc);

          await (this.service as any).yjsMysql.saveYjsDocument(roomName, roomName, snapshot);

          target.updates.length = 0; // 因为啥 直接获取内存中的 完整数据 ，所以清空其他的 增量数据的任务就行了！！

          target.lastFlush = Date.now();

        } catch (e) {
          this.app.logger.warn("[socket] flush failed room=%s err=%s", roomName, String(e));
        } finally {
          target.flushing = false;
          target.timer = null;
        }
      }, FLUSH_INTERVAL_MS);
    };

    // 安排一次对外广播的“短窗口合并”
    const scheduleBroadcast = (roomName: string) => {
      let buf = broadcastBuffers.get(roomName);
      if (!buf) {
        buf = { updates: [], timer: null };
        broadcastBuffers.set(roomName, buf);
      }
      if (buf.timer) return; // 已经安排，无需重复
      buf.timer = setTimeout(() => {
        const target = broadcastBuffers.get(roomName);
        if (!target || target.updates.length === 0) {
          if (target) target.timer = null;
          return;
        }
        try {
          // 合并窗口内多条增量为 1 条，再广播一次
          const merged = Y.mergeUpdates(target.updates);
          this.io?.to(roomName).emit("update", merged.buffer);
        } finally {
          target.updates.length = 0;
          target.timer = null;
        }
      }, BROADCAST_COALESCE_MS);
    };

    // 帮助函数：获取或初始化房间（首次会尝试从 MySQL 恢复）
    const getOrCreateRoom = async (roomName: string): Promise<RoomContext> => {
      const existed = rooms.get(roomName);
      if (existed) return existed;
      const doc = new Y.Doc();
      const awareness = new Awareness(doc);
      // 从 DB 恢复：如果有 yjs_state 则应用
      try {
        const state = await (this.service as any).yjsMysql.getYjsDocument(roomName, roomName);
        if (state) Y.applyUpdate(doc, state);
      } catch { }
      const ctx = { doc, awareness } as RoomContext;
      rooms.set(roomName, ctx);
      return ctx;
    };

    this.io = new SocketIOServer(server, {
      cors: {
        origin: ["http://localhost:5173", "http://localhost:5174", "http://127.0.0.1:5173", "http://localhost:3000"],
        methods: ["GET", "POST"],
        allowedHeaders: ["content-type", "cookie"],
        credentials: true
        // credentials: true 表示在跨域（CORS/Socket.IO）配置中允许请求携带“凭证”，如 Cookies、Authorization 头等。启用后服务器会返回 Access-Control-Allow-Credentials: true，客户端也需设置 withCredentials: true，且 origin 不能为 *。
      },
      transports: ["websocket", "polling"],
      allowEIO3: true,
      pingTimeout: 60000,
      pingInterval: 25000
    });
    appAny.__socketIO = this.io;

    // 单元测试环境：不对外监听端口，避免端口占用；仅构建事件与内存文档，满足 /socket/status 用例
    if (this.app.config.env === "unittest") {
      this.app.logger.info("[socket] unittest mode, skip listen()");
      return this.io;
    }

    this.io.on("connection", socket => {
      let currentRoom: string | null = null;
      let joinedProcessed = false;
      let syncStep1Processed = false; // 防止重复处理sync-step1

      // 解析当前连接的用户ID（仅通过 Cookie 明文 SID 校验）
      let currentUserId: string | null = null;
      const sessionId = this.getSessionIdFromSocket(socket);
      this.app.logger.info("[socket] sid=%s", sessionId || "(empty)");
      let resolveUser: Promise<void> | null = null;
      if (sessionId) {
        resolveUser = (async () => {
          currentUserId = await this.getUserIdBySessionId(sessionId);
          this.app.logger.info("[socket] uid=%s", currentUserId || "(null)");
        })();
      } else {
        this.app.logger.info("[socket] uid=%s", "(null)");
      }

      const processJoin = async (data: any) => {
        if (joinedProcessed) return;
        joinedProcessed = true;
        if (resolveUser) {
          try { await resolveUser; } catch (err) { this.app.logger.warn("[socket] resolve user failed id=%s err=%s", socket.id, String(err)); }
        }
        const { roomName } = data || {};
        if (!roomName) { socket.emit("error", { message: "房间名称不能为空" }); return; }
        currentRoom = roomName;
        socket.join(roomName);
        const ctx = await getOrCreateRoom(roomName);

        // 下发权限（从短期缓存读取，降低 DB 访问）
        const { effectivePermission, canWrite, isOwner } = await getPermissionCached(roomName, currentUserId);
        socket.emit("permission", { permission: effectivePermission, canWrite });
        this.app.logger.info("[socket] perm room=%s uid=%s perm=%s write=%s owner=%s", roomName, currentUserId || "(null)", effectivePermission, String(canWrite), String(isOwner));

        // 兼容：直接推送一次完整状态（便于无 state vector 的老客户端）
        // const full = Y.encodeStateAsUpdate(ctx.doc);
        // socket.emit("sync-doc", full.buffer);
        //
        // y-protocols：首次协商，客户端发送 state vector（sync-step1），服务端返回缺失 diff（sync-step2）

        // state vector 状态 向量
        // 防止重复处理sync-step1事件
        const handleSyncStep1 = (vectorBuf: ArrayBuffer) => {
          if (syncStep1Processed) {
            this.app.logger.warn("[sync-step1] 已处理过，忽略重复请求 room=%s", roomName);
            return;
          }
          syncStep1Processed = true;
          try {
            this.app.logger.info("[sync-step1] 触发sync-step1事件", {
              vectorBuf,
            });
            const vector = new Uint8Array(vectorBuf);
            const diff = Y.encodeStateAsUpdate(ctx.doc, vector);
            socket.emit("sync-step2", diff.buffer);
          } catch (e) {
            this.app.logger.warn("[socket] sync-step1 failed room=%s err=%s", roomName, String(e));
            syncStep1Processed = false; // 失败时重置状态，允许重试
          }
        };

        // 只监听一次，避免重复注册
        socket.removeAllListeners("sync-step1");
        socket.once("sync-step1", handleSyncStep1);

        // 增量更新：更新前实时确认写权限（使用短期缓存，避免高频 DB）
        socket.on("update", async (updateBuf: ArrayBuffer) => {
          // 仅记录字节大小，避免日志里输出二进制内容
          const size = (updateBuf as ArrayBuffer)?.byteLength || 0;
          // 降低热路径日志，避免高频 IO 拖慢实时性
          // this.app.logger.debug?.("[update] bytes=%d room=%s", size, roomName);
          // 写权限检查（缓存命中则跳过 DB）
          const { canWrite: allowWriteNow } = await getPermissionCached(roomName, currentUserId);
          if (!allowWriteNow) {
            // this.app.logger.warn("[socket] write denied room=%s uid=%s", roomName, currentUserId || "(null)");
            socket.emit("permission-denied", { code: "READ_ONLY", message: "当前为只读权限，禁止修改" });
            return;
          }
          const update = new Uint8Array(updateBuf);
          // 1) 服务端权威 Doc 先应用增量，确保后续快照基于最新状态
          Y.applyUpdate(ctx.doc, update);

          // 不然高并发下，会有很多 广播事件 来会触发。
          // 2) 短窗口合并对外广播（40ms 内多条合并为一次），减少广播风暴
          // 就是 延迟  this.io?.to(roomName).emit("update", merged.buffer);
          {
            let bbuf = broadcastBuffers.get(roomName);
            if (!bbuf) {
              bbuf = { updates: [], timer: null };
              broadcastBuffers.set(roomName, bbuf);
            }
            bbuf.updates.push(update);
            scheduleBroadcast(roomName);
          }

          // 3) 写入缓冲区，交由定时器或阈值触发进行“批量落库”
          // 获取文档对应的缓冲区
          let buf = updateBuffers.get(roomName);

          if (!buf) {
            buf = { updates: [], timer: null, lastFlush: 0, flushing: false };
            updateBuffers.set(roomName, buf);
          }

          // 将更新写入缓冲区
          buf.updates.push(update);

          if (buf.updates.length >= MAX_UPDATES_BEFORE_FLUSH) {
            // 超过阈值：尝试立即刷库（若未在刷新中）
            if (!buf.flushing) { // 如果缓冲区不在刷新中
              try {
                buf.timer && clearTimeout(buf.timer);
              } catch { }
              buf.timer = null;
              // 直接进行一次快照落库（与定时器 flush 逻辑一致）
              // 直接获取  当前内存 Doc 数据的完整状态。
              const snapshot = Y.encodeStateAsUpdate(ctx.doc);
              try {
                await (this.service as any).yjsMysql.saveYjsDocument(roomName, roomName, snapshot);
                buf.updates.length = 0;
                buf.lastFlush = Date.now();
              } catch (e) {
                this.app.logger.warn("[socket] immediate flush failed room=%s err=%s", roomName, String(e));
              }
            }

          } else {
            // 未超过阈值：安排一次“延迟落库”任务
            scheduleFlush(roomName, ctx);
          }

          // this.app.logger.info("[socket] update applied (buffered) room=%s bytes=%d pending=%d", roomName, size, buf.updates.length);
        });

        // awareness：直接转发（可演进为服务端维护权威状态）
        socket.on("awareness-update", (encoded: ArrayBuffer) => {
          socket.to(roomName).emit("awareness-update", encoded);
        });

        // 告知客户端已成功加入房间，便于客户端在收到该事件后再发送 sync-step1，避免竞态
        try {
          socket.emit("joined", { roomName });
        } catch { }
      };

      // 统一监听，提前捕获 joinRoom（避免连接早期丢事件）
      socket.onAny((event, ...args) => {
        if (event === "joinRoom" && !joinedProcessed) {
          void processJoin(args?.[0]);
        }
      });

      // 显式监听（常规路径）
      socket.on("joinRoom", async data => { await processJoin(data); });

      // 强制同步：客户端提交完整 state，与服务器状态合并后确认
      // 强制同步是“把客户端当前完整状态打包上送，服务端与自己的状态做一次合并并落库”的兜底通道。作用是快速把“客户端认为正确的最新内容”与“服务器/数据库里的旧内容”对齐，避免只靠大量增量回放或在异常时无法收敛。
      socket.on("force-sync", async (data: { roomName: string; state: ArrayBuffer }) => {
        const { roomName, state } = data || ({} as any);
        if (!roomName || !state) return;
        const ctx = currentRoom === roomName ? await getOrCreateRoom(roomName) : await getOrCreateRoom(roomName);
        const local = Y.encodeStateAsUpdate(ctx.doc);
        await (this.service as any).yjsMysql.mergeDocumentStates(roomName, roomName, [local, new Uint8Array(state)]);
        // 重新加载最新库态
        const merged = await (this.service as any).yjsMysql.getYjsDocument(roomName, roomName);
        if (merged) {
          const applyTo = new Y.Doc();
          Y.applyUpdate(applyTo, merged);
          // 用合并后的状态替换内存 doc
          ctx.doc.destroy();
          (ctx as any).doc = applyTo;
        }
        socket.emit("force-sync-complete");
      });

      socket.on("disconnect", reason => {
        this.app.logger.info("[socket] disconnected id=%s reason=%s", socket.id, reason);
        if (currentRoom) socket.leave(currentRoom);
      });
    });

    // 优先读取环境变量，测试场景可用 SOCKET_PORT=0 让系统分配端口
    let port = Number(process.env.SOCKET_PORT || 7002);
    const listen = (p: number) =>
      new Promise<void>((resolve, reject) => {
        const onError = (err: any) => {
          server.removeListener("listening", onListening);
          reject(err);
        };
        const onListening = () => {
          server.removeListener("error", onError);
          const actual = (server.address() as any)?.port || p;
          this.app.logger.info("[socket] listening on %d", actual);
          resolve();
        };
        server.once("error", onError);
        server.once("listening", onListening);
        // 绑定到本地回环减少 CI/单测与系统端口冲突
        server.listen(p, "127.0.0.1");
      });

    try {
      await listen(port);
    } catch (e: any) {
      if (e && e.code === "EADDRINUSE") {
        // 端口占用时回退到 0（随机可用端口），避免单测冲突
        this.app.logger.warn("[socket] port %d in use, fallback to 0", port);
        port = 0;
        await listen(port);
      } else {
        throw e;
      }
    }

    return this.io;
  }

  public getIO(): SocketIOServer | null {
    return ((this.app as any).__socketIO as SocketIOServer) || this.io;
  }
}


